package org.zjvis.datascience.service.dataset;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.zjvis.datascience.common.dto.DatasetDTO;
import org.zjvis.datascience.common.dto.GreenPlumDTO;
import org.zjvis.datascience.common.dto.dataset.DatasetColumnDTO;
import org.zjvis.datascience.common.dto.datax.*;
import org.zjvis.datascience.common.util.RedisUtil;
import org.zjvis.datascience.service.dataset.quartz.SpringContextJobUtil;

import java.util.*;

/**
 * @description DataX相关工具
 * @date 2021-12-06
 */
public class DataXJsonBuilder {

    public static String buildDataXJson(DatasetDTO datasetDTO, GreenPlumDTO greenPlumDTO) {
        JSONObject dataJson = JSON.parseObject(datasetDTO.getDataJson());
        String userId = datasetDTO.getUserId().toString();
        JSONObject dataConfig = null;
        try {
            dataConfig = JSON.parseObject(DataConfigEncryption.decrypt(userId, datasetDTO.getDataConfig()));
        } catch (Exception e) {
            e.printStackTrace();
        }
        JSONObject incrementDataConfig = JSON.parseObject(datasetDTO.getIncrementalDataConfig());
        Content content = new Content(buildReader(dataJson, dataConfig, incrementDataConfig),
                buildWriter(dataJson, greenPlumDTO));
        Map<String, Map<String, Integer>> setting = new HashMap<>();
        Map<String, Integer> speed = new HashMap<>();
        speed.put("channel", 8);
        setting.put("speed", speed);
        Job job = new Job(setting, new ArrayList<>(Collections.singletonList(content)));
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("job", job);
        return JSON.toJSONString(jsonObject);
    }

    public static DataConfig buildReader(JSONObject dataJson, JSONObject dataConfig,
                                         JSONObject incrementDataConfig) {
        List<Connection> connList = new ArrayList<>();
        ReaderConnection connection = new ReaderConnection();
        connection.setTable(new ArrayList<>(Collections.singletonList(dataConfig.getString("tableName"))));
        connection.setJdbcUrl(new ArrayList<>(Collections.singletonList(dataConfig.getString("url"))));
        connList.add(connection);
        List<DatasetColumnDTO> datasetColumnDTOS = JSONArray.parseArray(dataJson.getString("columnMessage"), DatasetColumnDTO.class);
        List<String> columns = new ArrayList<>();
        for (DatasetColumnDTO datasetColumnDTO : datasetColumnDTOS) {
            columns.add("`" + datasetColumnDTO.getName() + "`");
        }
        ReaderParameter parameter = new ReaderParameter();
        parameter.setUsername(dataConfig.getString("user"));
        parameter.setPassword(dataConfig.getString("password"));
        parameter.setColumn(columns);
        parameter.setConnection(connList);
        String incrementColumn = incrementDataConfig.getString("incrementColumn");
        String lastValue = incrementDataConfig.getString("lastValue");
        if (lastValue.length() == 0) {
            parameter.setWhere("1=1");
        } else {
            Double value = null;
            StringBuilder where = new StringBuilder(incrementColumn + ">");
            try {
                value = Double.parseDouble(lastValue);
            } catch (Exception ignored) {

            }
            if (value == null) {
                where.append("'").append(lastValue).append("'");
            } else {
                where.append(value);
            }
            parameter.setWhere(where.toString());
        }
        RedisUtil redisUtil = (RedisUtil) SpringContextJobUtil.getBean("redisUtil");
        redisUtil.set("lastValue:" + dataConfig.getString("tableName"), lastValue);
        return new DataConfig(dataConfig.getString("databaseType") + "reader", parameter);
    }

    public static DataConfig buildWriter(JSONObject dataJson, GreenPlumDTO greenPlumDTO) {
        List<Connection> connList = new ArrayList<>();
        WriterConnection connection = new WriterConnection();
        connection.setTable(new ArrayList<>(Collections.singletonList("dataset." + dataJson.getString("table"))));
        connection.setJdbcUrl(greenPlumDTO.getUrl());
        connList.add(connection);
        List<DatasetColumnDTO> datasetColumnDTOS = JSONArray.parseArray(dataJson.getString("columnMessage"), DatasetColumnDTO.class);
        List<String> columns = new ArrayList<>();
        for (DatasetColumnDTO datasetColumnDTO : datasetColumnDTOS) {
            columns.add(datasetColumnDTO.getName());
        }
        Parameter parameter = new Parameter();
        parameter.setUsername(greenPlumDTO.getUserName());
        parameter.setPassword(greenPlumDTO.getPassword());
        parameter.setColumn(columns);
        parameter.setConnection(connList);
        return new DataConfig("postgresqlwriter", parameter);
    }
}
